package com.smile.cloud.order.service;

import jodd.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 订单事务生产者
 *
 * @author LGC
 */
@Slf4j
@Component
public class TransactionProducer {

    private final static String ORDER_TRANS_GROUP = "order_trans_group";
    private TransactionMQProducer producer;

    //用于执行本地事务和事务状态回查的监听器
    @Resource
    private OrderTransactionListener orderTransactionListener;
    //执行任务的线程池
    ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10,
            60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(50),
            ThreadFactoryBuilder.create().setNameFormat("trans-msg-task-%d").get());

    @PostConstruct
    public void init() {
        log.info(ORDER_TRANS_GROUP + " producer 正在创建---------------------------------------");
        producer = new TransactionMQProducer(ORDER_TRANS_GROUP);
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setSendMsgTimeout(Integer.MAX_VALUE);
        producer.setExecutorService(executor);
        producer.setTransactionListener(orderTransactionListener);
        this.start();
    }

    private void start() {
        try {
            this.producer.start();
            log.info(ORDER_TRANS_GROUP + " producer server 开启成功----------------------------------");
        } catch (MQClientException e) {
            log.info(ORDER_TRANS_GROUP + " producer server 开启异常：{0}----------------------------------", e);
        }
    }

    //事务消息发送
    public TransactionSendResult send(String data, String topic) throws MQClientException {
        Message message = new Message(topic, data.getBytes());
        return this.producer.sendMessageInTransaction(message, null);
    }
}
